package com.ods.mq.controller;

import com.alibaba.fastjson.JSONObject;
import com.amych.base.message.ResponseMessage;
import com.amych.base.message.resolver.RequestData;
import com.ods.mq.db.dto.KafkaDTO;
import com.ods.mq.db.entity.KafkaEntity;
import com.ods.mq.db.service.KafkaService;
import com.ods.mq.tool.KafkaTool;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @Description
 * @Author YuanX
 * @Date 2021/10/20
 * @Param
 * @return
 **/
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Autowired
    private KafkaService _kafkaService;

    /**
     * @return com.amych.base.message.ResponseMessage
     * @Description 获取浏览器访问地址中的ip和端口
     * @Param [request]
     **/
    @PostMapping("/reqAddress")
    public ResponseMessage reqAddress(HttpServletRequest request) {
        return ResponseMessage.ok(request.getServerName() + ":" + request.getServerPort());
    }

    /**
     * @return ResponseDto
     * @Description 获取所有哦主题
     * @Param [sourceId]
     **/
    @PostMapping("/getTopics")
    public ResponseMessage getTopics(@RequestData(value = "data.id", required = true) String id,
                                     @RequestData(value = "data.name", required = false) String name) {
        KafkaEntity kafkaEntity = _kafkaService.getById(id);
        return KafkaTool.listTopicsWithOptions(kafkaEntity.getBroker(), name);
    }

    /**
     * 创建主题
     * @param dto
     * @return
     * @throws Exception
     */
    @PostMapping("/addTopic")
    public ResponseMessage addTopic(
            @RequestData(value = "data") KafkaDTO dto) throws Exception {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        return KafkaTool.createTopic(kafkaEntity.getBroker(), dto.getName(), dto.getPartition(), dto.getReplica());
    }


    /**
     * @return boolean
     * @Description 删除主题
     * @Param [sourceId, topic]
     **/
    @PostMapping("/findTopic")
    public ResponseMessage findTopic(@RequestData(value = "data") KafkaDTO dto) {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        return KafkaTool.delTopic(kafkaEntity.getBroker(), dto.getName());
    }

    /**
     * @return com.amych.base.message.ResponseMessage
     * @Description 查找主题
     * @Param [dto]
     **/
//    @PostMapping("/findTopics")
//    public ResponseMessage findTopics(@RequestData(value = "data") KafkaDTO dto) {
//        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
//        return KafkaTool.listTopicsWithOptions(kafkaEntity.getBroker(), dto.getName());
//    }

    /**
     * @return com.alibaba.fastjson.JSONObject
     * @Description 主题详情
     * @Param [dto]
     **/
    @PostMapping("/detailTopic")
    public ResponseMessage detailTopic(@RequestData(value = "data") KafkaDTO dto) throws Exception {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        return ResponseMessage.ok().putData(KafkaTool.topicInfo(kafkaEntity.getBroker(), kafkaEntity.getName()));
    }

    /**
     * kafka消息生产
     *
     * @param dto
     * @return
     */
    @PostMapping("/produce")
    public ResponseMessage produce(@RequestData(value = "data") KafkaDTO dto) throws ExecutionException, InterruptedException {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        Producer<String, String> producer = KafkaTool.producerConfig(kafkaEntity.getBroker());
        if (dto.getBatch()) {
            String[] messages = dto.getMessage().split("\n");
            for (String ms : messages) {
                Future<RecordMetadata> send = producer.send(new ProducerRecord<>(dto.getName(), ms));
                send.get();
            }
        } else {
            Future<RecordMetadata> send = producer.send(new ProducerRecord<>(dto.getName(), dto.getMessage()));
            send.get();
        }
        producer.close();
        return ResponseMessage.ok("消息生产成功！");
    }

    /**
     * 获取所有数据源 *
     *
     * @return
     */
    @PostMapping("/dataSource")
    public ResponseMessage dataSource() {
        return ResponseMessage.ok().putData(_kafkaService.list());
    }

    /**
     * 删除数据源 *
     *
     * @return
     */
    @PostMapping("/delSource")
    public ResponseMessage delSource(@RequestData(value = "data") KafkaDTO dto) {
        return ResponseMessage.ok().putData(_kafkaService.removeById(dto.getId()));
    }

    /**
     * 添加数据源
     * @param dto
     * @return
     */
    @PostMapping("/addSource")
    public ResponseMessage addSource(@RequestData(value = "data") KafkaDTO dto) {
        KafkaEntity entity = new KafkaEntity();
        BeanUtils.copyProperties(dto, entity);
        return ResponseMessage.ok().putData(_kafkaService.saveOrUpdate(entity));
    }

    /**
     * 获取集群列表
     * @param dto
     * @return
     */
    @PostMapping("/listCluster")
    public ResponseMessage listCluster(@RequestData(value = "data") KafkaDTO dto) {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        return KafkaTool.clusterInfo(kafkaEntity.getBroker());
    }

    /**
     * group  列表
     * @param dto
     * @return
     */
    @PostMapping("/findGroup")
    public ResponseMessage findGroup(@RequestData(value = "data") KafkaDTO dto) {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        return KafkaTool.groupInfo(kafkaEntity.getBroker(), null);
    }

    /**
     * group  group详情
     * @param dto
     * @return
     */
    @PostMapping("/detailGroup")
    public ResponseMessage detailGroup(@RequestData(value = "data") KafkaDTO dto) {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        return KafkaTool.groupInfo(kafkaEntity.getBroker(), dto.getName());
    }



    /**
     *
     * 删除group
     * @param dto
     * @return
     */
    @PostMapping("/delGroup")
    public ResponseMessage deleteGroup(@RequestData(value = "data") KafkaDTO dto) {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        return KafkaTool.delGroup(kafkaEntity.getBroker(), kafkaEntity.getName());
    }

    /**
     * 主题下的所有 group
     * @param dto
     * @return
     */
    @PostMapping("/topicGroup")
    public ResponseMessage topicGroup(@RequestData(value = "data") KafkaDTO dto) {
        KafkaEntity kafkaEntity = _kafkaService.getById(dto.getId());
        return KafkaTool.topicAllGroup(kafkaEntity.getBroker(), dto.getName());
    }


}
